package consistentHash;

import sun.security.provider.MD5;

import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.Collection;
import java.util.Iterator;
import java.util.SortedMap;
import java.util.TreeMap;

/**
 * @author 03010570
 * @date 2020/07/07
 * describe:    一致性 Hash 算法
 */
public class ConsistentHashRouter {

    /**
     * hashCode -> 虚拟节点
     */
    private SortedMap<Long, VirtualNode> ring = new TreeMap<Long, VirtualNode>();
     MD5Hash hashFunction = new MD5Hash();

    public ConsistentHashRouter(Collection<PhysicalNode> physicalNodes,Integer vnodeCount) {
        for (PhysicalNode physicalNode : physicalNodes) {
            addPhysicalNode(physicalNode,vnodeCount);
        }
    }

    /**
     * 新增物理节点
     * @param node
     */
    public void addPhysicalNode(PhysicalNode node,Integer vnodeCount){
        int existingReplicas = getReplicas(node.toString());
        for(int i = 0;i<vnodeCount;i++){
            VirtualNode virtualNode = new VirtualNode(node, i + existingReplicas);
            ring.put(hashFunction.hash(virtualNode.toString()),virtualNode);
        }
    }

    /**
     * 删除物理节点
     * @param node
     */
    public void removePhysicalNode(PhysicalNode node){
        Iterator<Long> it = ring.keySet().iterator();
        while(it.hasNext()){
            Long key = it.next();
            VirtualNode virtualNode = ring.get(key);
            if(virtualNode.matches(node.toString())){
                it.remove();
            }
        }
    }

    /**
     * 根据 节点名称 来查找 物理节点
     * @param key
     * @return
     */
    public PhysicalNode getPhysicalNodeByKey(String key){
        if(ring.isEmpty()){
            return null;
        }
        Long hashKey = hashFunction.hash(key);
        SortedMap<Long, VirtualNode> tailMap = ring.tailMap(hashKey);
        hashKey = tailMap != null && !tailMap.isEmpty() ? tailMap.firstKey() : ring.firstKey();
        return ring.get(hashKey).getPhysicalNode();
    }

    /**
     * 根据物理节点名称来获取 虚拟节点的下标计数
     * @param nodeName
     * @return
     */
    public  int getReplicas(String nodeName){
        int index = 0;
        for (VirtualNode virtualNode : ring.values()) {
            if(virtualNode.matches(nodeName)){
                index++;
            }
        }
        return index;
    }



    /**
     * 静态内部类构建 hash
     */
    static class MD5Hash {
        MessageDigest instance;

        public MD5Hash() {
            try {
                instance = MessageDigest.getInstance("MD5");
            } catch (NoSuchAlgorithmException e) {
                e.printStackTrace();
            }
        }

        /**
         * 获取 hash 值
         * @param key
         * @return
         */
        Long hash(String key) {
            instance.reset();
            instance.update(key.getBytes());
            byte[] digest = instance.digest();

            long h = 0;
            for (int i = 0; i < 4; i++) {
                h <<= 8;
                h |= ((int) digest[i]) & 0xFF;
            }
            System.out.println(h);
            return h;
        }

    }
}
